package com.ferry.config.consumer;

import com.ferry.registry.domain.registry.RegistryType;
import com.ferry.registry.service.IRegistryService;
import com.ferry.registry.zk.ZkRegistryServiceImpl;
import com.ferry.serial.api.SerializerType;
import com.ferry.transport.api.Connection;
import com.ferry.transport.api.Directory;
import com.ferry.transport.api.config.ChildConfig;
import com.ferry.transport.api.config.TcpConfig;
import com.ferry.transport.netty.client.ConsumerClient;
import com.ferry.transport.netty.client.DefaultConsumerClient;
import com.ferry.transport.netty.consumer.*;
import com.ferry.transport.netty.proxy.ProxyFactory;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import java.lang.reflect.Field;

/**
 * Created by daiyong
 */
public class FerryClientConfig<T> implements ApplicationContextAware, BeanPostProcessor {

	private ConsumerClient consumerClient;

	private IRegistryService registryService;

	private String appName;

	//注册中心地址
	private String registryAddress;

	//配置
	private ChildConfig childConfig = null;

	//连接器
	private Connector<Connection> connector;

	@Override
	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {

		//链接zk
		connectRegistry(registryAddress);

		//创建connector
		connector = new DefaultNettyTcpConnector();

		consumerClient = new DefaultConsumerClient(registryService, appName, connector);

		//配置网络层参数
		if (null != childConfig) {
			TcpConfig config = connector.config();
			config = childConfig;
		}

		Runtime.getRuntime().addShutdownHook(new Thread(() -> consumerClient.shutdownGracefully()));
	}

	private void connectRegistry(String registryAddress) {
		//默认只有zk注册中心，以后可能回增加etcd
		RegistryType registryType = getRegistryType(registryAddress);
		if (null == registryType) {
			throw new IllegalArgumentException("不支持当前类型注册中心");
		}
		if (registryType == RegistryType.ZOOKEEPER) {
			registryService = buildZkRegistryService(registryAddress);
		}
	}

	private RegistryType getRegistryType(String registryAddress) {

		String[] split = registryAddress.split("//");
		String protocol = split[0].replace(":", "");

		return RegistryType.getRegistryTypeByProtocol(protocol);
	}

	private IRegistryService buildZkRegistryService(String registryAddress) {
		IRegistryService registryService = new ZkRegistryServiceImpl();
		registryService.connect(registryAddress.split("//")[1]);
		return registryService;
	}

	@Override
	public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
		return bean;
	}

	@Override
	public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {

		Field[] fields = bean.getClass().getDeclaredFields();
		for (Field field : fields) {
			field.setAccessible(true);
			if (field.isAnnotationPresent(FerryConsumer.class)) {
				FerryConsumer annotation = field.getAnnotation(FerryConsumer.class);

				//创建并且管理连接
				Directory directory = new Directory();
				directory.setGroup(annotation.group());
				String providerName = "";
				providerName = StringUtils.isNotBlank(annotation.providerName()) ? annotation.providerName() : field.getType().getSimpleName();
				directory.setServiceProviderName(providerName);
				directory.setVersion(annotation.version());

				ConnectionWatcher connectionWatcher = consumerClient.watchConnections(directory);
				if (!connectionWatcher.waitForAvailable(3000)) {
					throw new RuntimeException("connect timeout");
				}

				//构造代理对象
				ProxyFactory<T> proxyFactory = new ProxyFactory<>();
				Class<?> declaringClass = field.getType();
				proxyFactory.interfaceClass((Class<T>) declaringClass);
				proxyFactory.group(annotation.group());
				proxyFactory.providerName(providerName);
				proxyFactory.version(annotation.version());
				proxyFactory.serializerType(SerializerType.parse(annotation.serializerType()));
				proxyFactory.loadBalancerType(LbType.parse(annotation.LbType()));
				proxyFactory.invokeType(InvokeType.parse(annotation.InvokeType()));
				proxyFactory.timeout(annotation.timeout());
				proxyFactory.failStrategy(FailStrategy.parse(annotation.failStrategy()));
				proxyFactory.failoverRetries(annotation.failoverRetries());
				proxyFactory.connector(this.connector);
				proxyFactory.appName(appName);

				Object proxyInstance = proxyFactory.newProxyInstance();
				//重写filed
				try {
					field.set(bean, proxyInstance);
				} catch (IllegalAccessException e) {
					throw new RuntimeException(e);
				}
			}
		}

		return bean;
	}

	public String getAppName() {
		return appName;
	}

	public void setAppName(String appName) {
		this.appName = appName;
	}

	public String getRegistryAddress() {
		return registryAddress;
	}

	public void setRegistryAddress(String registryAddress) {
		this.registryAddress = registryAddress;
	}

	public ChildConfig getChildConfig() {
		return childConfig;
	}

	public void setChildConfig(ChildConfig childConfig) {
		this.childConfig = childConfig;
	}
}
